package com.nepu.spark.realtime.app

import java.time.{LocalDate, Period}
import java.util

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.nepu.spark.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.nepu.spark.realtime.util.{KafkaOffsetUtils, MyElasticSearchUtils, MyKafkaUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
  * 这里处理的是订单的宽表
  *
  * @author chenshuaijun
  * @create 2022-11-01 16:15
  */

/**
  * 订单宽表的处理的流程
  * 1、首先是创建sparkStreaming的上下文的环境对象
  * 2、提取offset
  * 3、从kafka中消费数据
  * 4、提取offset
  * 5、数据处理
  *   5.1、转换数据结构
  *   5.2、进行维度的关联
  *   5.3、双流join
  * 6、数据写入到es
  * 7、提交offset
  */
object DwdOrderWideTable {
  def main(args: Array[String]): Unit = {
    // 1、准备上下文的环境对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("dwd_order_width_table")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 定义topic
    val orderInfoTopicName: String = "DWD_ORDER_INFO_I"
    val orderInfoGroup: String = "ORDER_INFO_GROUP"
    val orderDetailTopicName: String = "DWD_ORDER_DETAIL_I"
    val orderDetailGroup: String = "ORDER_DETAIL_GROUP"
    // 2、从kafka中消费数据
    // 2.1 消费order_info的数据
    val orderInfoOffset: Map[TopicPartition, Long] = KafkaOffsetUtils.getOffset(orderInfoTopicName, orderInfoGroup)
    var orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderInfoOffset != null && orderInfoOffset.nonEmpty) {
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, orderInfoTopicName, orderInfoGroup, orderInfoOffset)
    } else {
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, orderInfoTopicName, orderInfoGroup)
    }
    // 2.2 消费order_detail中的数据
    val orderDetailOffset: Map[TopicPartition, Long] = KafkaOffsetUtils.getOffset(orderDetailTopicName, orderDetailGroup)
    var orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderDetailOffset != null && orderDetailOffset.nonEmpty) {
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, orderDetailTopicName, orderDetailGroup, orderDetailOffset)
    } else {
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, orderDetailTopicName, orderDetailGroup)
    }

    // 提取offset
    var orderInfoOffsetRanges: Array[OffsetRange] = null
    val orderInfoOffsetRangesDStream: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform(
      rdd => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    var orderDetailOffsetRanges: Array[OffsetRange] = null
    val orderDetailOffsetRangesDStream: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform(
      rdd => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    // 转换数据结构
    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetRangesDStream.map(
      consumerRecord => {
        val orderInfoJson: String = consumerRecord.value()
        val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
        orderInfo
      }
    )

    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetRangesDStream.map(
      consumerRecord => {
        val orderDetailJson: String = consumerRecord.value()
        val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
        orderDetail
      }
    )

    // 对orderInfo进行维度关联
    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      iterator => {
        // 获取jedis的连接
        val jedis: Jedis = MyRedisUtils.getJedisClient()
        val orderInfos = new ListBuffer[OrderInfo]()
        for (orderInfo <- iterator) {
          // 用户维度的关联
          val userId: Long = orderInfo.user_id
          // 查询key的封装
          val userRedisKey: String = s"DIM:USER_INFO:$userId"
          // 从Redis中获取数据
          val userInfo: String = jedis.get(userRedisKey)
          // 转换成Json的对象
          val userInfoObj: JSONObject = JSON.parseObject(userInfo)
          // 拿到性别信息
          val userGender: String = userInfoObj.getString("gender")
          // 拿到出生日期
          val birthday: String = userInfoObj.getString("birthday")
          val birthdayDate: LocalDate = LocalDate.parse(birthday)
          val currentDate: LocalDate = LocalDate.now()
          val userAge: Int = Period.between(birthdayDate, currentDate).getYears
          orderInfo.user_gender = userGender
          orderInfo.user_age = userAge

          // 地区维度数据关联
          val provinceId: Long = orderInfo.province_id
          // 查询key的封装
          val provinceRedisKey: String = s"DIM:BASE_PROVINCE:$provinceId"
          // 到Redis中查询地区的数据
          val provinceInfoJson: String = jedis.get(provinceRedisKey)
          // 转换成Json的对象
          val provinceInfo: JSONObject = JSON.parseObject(provinceInfoJson)
          // 获取省份的名字
          val provinceName: String = provinceInfo.getString("name")
          // 获取地区编码
          val provinceAreaCode: String = provinceInfo.getString("area_code")
          // 获取3166编码
          val provinceIso3166: String = provinceInfo.getString("iso_3166_2")
          // 获取iso_code
          val provinceIsoCode: String = provinceInfo.getString("iso_code")
          // 封装数据
          orderInfo.province_name = provinceName
          orderInfo.province_area_code = provinceAreaCode
          orderInfo.province_3166_2_code = provinceIso3166
          orderInfo.province_iso_code = provinceIsoCode

          // 创建时间和小时的封装
          val orderCreateTime: String = orderInfo.create_time
          val orderCreateTimeSplit: Array[String] = orderCreateTime.split(" ")
          val createDate: String = orderCreateTimeSplit(0)
          val createHour: String = orderCreateTimeSplit(1).split(":")(0)
          orderInfo.create_date = createDate
          orderInfo.create_hour = createHour
          orderInfos.append(orderInfo)
        }
        // 关闭jedis的连接
        jedis.close()
        orderInfos.toIterator
      }
    )

    // 进行双流的join
    /**
      * 对于双流的join这里会遇到如下的问题
      * 1、首先我们要使用的join类型：内连接（join）、左外连接(leftOuterJoin)、右外连接(rightOuterJoin)、全外连接(fullOuterJoin)
      * 如果要进行join的选择，我们需要先知道各个join的特征：
      * 1、内连接（join）: join的结果，之会保留两个流能够join的数据，对于那些无法join的数据，会全部的丢弃掉
      * 2、左外联 (leftOuterJoin)：左外联的思想是，我们想要把左表作为我们的主表，这样无论后期的join操作，
      * 左表中的数据有没有和右表join上，左表的数据都会完全的保留
      * 3、右外联（RightOuterJoin）：右外联的思想是，我们想要把右表作为我们的主表，这样无论后期的join操作，
      * 右表中的数据有没有和左表join上，右表的数据都会完全的保留
      * 4、全外联（FullOuterJoin）: join的结果中无论左右表的数据是否能够成功的被join上，这两个表的数据都会被保留下来
      * 2、我们最终选择的是全外联，主要的原因是，每产生一个订单数据，就必然会有一个订单详情数据，所以这两个流在理论上是一定可以Join成功的
      * 但是因为我们处理的是流数据，所以会出现，数据到达的时间是不一致的，即使是这样，我们仍旧是不能够丢弃我们的数据，所以只有使用全外联才能
      * 保证我们不会丢数距。
      */
    // join的数据是必须有key的，所以我们需要转化一下数据结构
    val orderWideMapDStream: DStream[(Long, OrderInfo)] = orderInfoDimDStream.map(orderInfo => (orderInfo.id, orderInfo))
    val orderDetailMapDStream: DStream[(Long, OrderDetail)] = orderDetailDStream.map(orderDetailInfo => (orderDetailInfo.order_id, orderDetailInfo))
    val orderFullOuterJoinDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderWideMapDStream.fullOuterJoin(orderDetailMapDStream)

    // 开始处理join失败的数据，我们解决的办法是，无论orderInfo中的数据有没有被join成功，他都会被写入到缓存中，对于orderDetail的数据，
    // 如果它被join成功，就就直接的丢掉，如果没有join成功，就会被写入到缓存中
    val orderWideDStream: DStream[OrderWide] = orderFullOuterJoinDStream.mapPartitions(
      orderFullIterable => {
        val orderWides = new ListBuffer[OrderWide]()
        val jedis: Jedis = MyRedisUtils.getJedisClient()
        for ((key, (orderInfoOp, orderDetailOp)) <- orderFullIterable) {
          // 我们双流join之后可能有如下的结果
          // 1、orderInfo有，orderDetail有
          // 2、orderInfo有，orderDetail没有
          // 3、orderInfo没有，orderDetail有
          // 对于1和2这种情况，我们都需要将orderInfo的数据缓存起来，所以这两个可以合并，一起处理
          // 对于3我们需要单独的进行处理
          if (orderInfoOp.isDefined) {
            // 取出orderInfo
            val orderInfo: OrderInfo = orderInfoOp.get
            if (orderDetailOp.isDefined) {
              // 取出orderDetail
              val orderDetail: OrderDetail = orderDetailOp.get
              val orderWide = new OrderWide(orderInfo, orderDetail)
              // 存入到orderWides中
              orderWides.append(orderWide)
            }
            // 将orderInfo缓存到Redis中
            // 拼接key
            val orderInfoKey: String = s"ORDER_JOIN:ORDER_INFO:${orderInfo.id}"
            jedis.setex(orderInfoKey, 24 * 3600, JSON.toJSONString(orderInfo, new SerializeConfig(true)))

            // 从Redis中读取缓存
            val orderDetailKey: String = s"ORDER_JOIN:ORDER_DETAIL:${orderInfo.id}"
            val orderDetails: util.Set[String] = jedis.smembers(orderDetailKey)
            if (orderDetails != null && orderDetails.size() > 0) {
              import scala.collection.JavaConverters._
              for (orderDetailJson <- orderDetails.asScala) {
                val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
                val orderWide = new OrderWide(orderInfo, orderDetail)
                orderWides.append(orderWide)
              }
            }
          } else {
            // 先取出orderDetail
            val orderDetail: OrderDetail = orderDetailOp.get
            // 这里处理的是orderInfo没有的情况
            val orderInfoKey: String = s"ORDER_JOIN:ORDER_INFO:${orderDetail.order_id}"
            // 先从Redis中读取orderInfo
            val orderInfoJson: String = jedis.get(orderInfoKey)
            if (orderInfoJson != null) {
              // 如果能取到对应的orderInfo的数据，就直接进行处理
              // 转换数据结构
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            } else {
              // 如果取不到orderInfo的数据，就将自己缓存起来
              val orderDetailKey: String = s"ORDER_JOIN:ORDER_DETAIL:${orderDetail.order_id}"
              jedis.sadd(orderDetailKey, JSON.toJSONString(orderDetail, new SerializeConfig(true)))
              jedis.expire(orderDetailKey, 24 * 3600)
            }

          }
        }
        jedis.close()
        orderWides.toIterator
      }
    )
    orderWideDStream.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          rddIter => {
            val orderWides: List[(String, OrderWide)] = rddIter.map(orderWide => (orderWide.detail_id.toString, orderWide)).toList
            if (orderWides != null && orderWides.nonEmpty) {
              // 提取时间
              val head: (String, OrderWide) = orderWides.head
              val date: String = head._2.create_date
              // 拼接es的索引
              val esIndex: String = s"gmall_order_wide_$date"
              // 将数据写入到es
              MyElasticSearchUtils.batchIdempotentWrite(esIndex, orderWides)
            }
          }
        )
        // 提交offset
        KafkaOffsetUtils.saveOffset(orderInfoTopicName, orderInfoGroup, orderInfoOffsetRanges)
        KafkaOffsetUtils.saveOffset(orderDetailTopicName, orderDetailGroup, orderDetailOffsetRanges)
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }


}
